package cn.itcast.flink.base;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Random;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.stream.Collectors;

/**
 * Author itcast
 * Date 2021/8/2 10:46
 * 实时（每秒钟）统计当天到当前时间的根据不同分类的订单的 top3 统计及最新的总金额
 * 步骤：
 * 1. 数据源 分析数据格式
 * 2. 数据结果  最终要分析成数据结构
 */
public class DoubleElevenBigScreen {
    public static void main(String[] args) throws Exception {
        //1. 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置执行的并发为 1 ，便于观察
        env.setParallelism(1);
        //2. 创建数据源 自定义 Source
        DataStreamSource<Tuple2<String/**分类**/, Double/**金额**/>> source = env.addSource(new MySource());
        //3. transformation
        //分组操作
        source.keyBy(t -> t.f0)
                //3.1 定义大小为一天的窗口，第二个参数表示中国使用的UTC+08:00时区比UTC时间早
                .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))
                //3.2定义一个1s的触发器
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
                //上面的3.1和3.2相当于自定义窗口的长度和触发时机
                //3.3聚合结果.aggregate(new PriceAggregate(), new WindowResult());
                .aggregate(new PriceAggregate(), new WindowResult())
                //以前的写法用的默认的聚合和收集
                //现在可以自定义如何对price进行聚合,并自定义聚合结果用怎样的格式进行收集
                //3.4看一下初步聚合的
                //每秒钟更新一次统计结果
                //Time size 为1s,表示计算最近1s的数据
                .keyBy(t -> t.getDateTime())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                //实现WindowResultProcess继承ProcessWindowFunction中实现该复杂业务逻辑,一次性将需求1和2搞定
                .process(new WindowResultProcess())
                .print();
        //-1.实时计算出11月11日00:00:00零点开始截止到当前时间的销售总额
        //-2.计算出各个分类的销售额top3
        //-3.每1秒钟更新一次统计结果
        //5.execute
        env.execute();

    }

    /**
     * 自定义数据源实时产生订单数据 Tuple2<分类, 金额>
     */
    public static class MySource implements SourceFunction<Tuple2<String, Double>> {
        private boolean flag = true;
        private String[] categorys = {"女装", "男装", "图书", "家电", "洗护", "美妆", "运动", "游戏", "户外", "家具", "乐器", "办公"};
        private Random random = new Random();

        @Override
        public void run(SourceContext<Tuple2<String, Double>> ctx) throws Exception {
            while (flag) {
                //随机生成分类和金额
                int index = random.nextInt(categorys.length);//[0~length) ==> [0~length-1]
                String category = categorys[index];//获取的随机分类
                double price = random.nextDouble() * 100;//注意nextDouble生成的是[0~1)之间的随机数,*100之后表示[0~100)
                ctx.collect(Tuple2.of(category, price));
                Thread.sleep(20);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }

    /**
     * 自定义价格聚合函数,其实就是对price的简单sum操作
     * AggregateFunction<IN, ACC, OUT>
     * AggregateFunction<Tuple2<String, Double>, Double, Double>
     */

    /**
     * 用于存储聚合的结果
     */
    /**
     * 用于存储聚合的结果
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class CategoryPojo {
        private String category;//分类名称
        private double totalPrice;//该分类总销售额
        private String dateTime;// 截止到当前时间的时间,本来应该是EventTime,但是我们这里简化了直接用当前系统时间即可
    }

    /**
     * * @param <IN>  The type of the values that are aggregated (input values) [男装,100] [男装,200] [女装,130] [男装,120]
     * * @param <ACC> The type of the accumulator (intermediate aggregate state).  [男装,300]
     * * @param <OUT> The type of the aggregated result
     */
    private static class PriceAggregate implements AggregateFunction<Tuple2<String, Double>, Double, Double> {
        @Override
        public Double createAccumulator() {
            return 0D;
        }

        @Override
        public Double add(Tuple2<String, Double> value, Double accumulator) {
            return value.f1 + accumulator;
        }

        @Override
        public Double getResult(Double accumulator) {
            return accumulator;
        }

        @Override
        public Double merge(Double a, Double b) {
            return a + b;
        }
    }

    /**
     * 自定义WindowFunction,实现如何收集窗口结果数据
     * interface WindowFunction<IN, OUT, KEY, W extends Window>
     * interface WindowFunction<Double, CategoryPojo, Tuple的真实类型就是String就是分类, W extends Window>
     * "yyyy-MM-dd HH:mm:ss"
     */
    private static class WindowResult implements WindowFunction<Double, CategoryPojo, String, TimeWindow> {
        @Override
        public void apply(String category, TimeWindow window, Iterable<Double> input, Collector<CategoryPojo> out) throws Exception {
            CategoryPojo categoryPojo = new CategoryPojo();
            //定义时间格式
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            //将分类传递进来
            categoryPojo.setCategory(category);
            //获取当前的处理时间
            long currentTimeMillis = System.currentTimeMillis();
            categoryPojo.setDateTime(sdf.format(currentTimeMillis));
            //将当前窗口内的总金额
            Double totalMoney = input.iterator().next();
            //将总金额进行四舍五入并且为了能够对大的数据进行计算，使用 BigDecimal 类
            BigDecimal bigDecimal = new BigDecimal(totalMoney);
            //进行四舍五入
            double roundTotalMoney = bigDecimal.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
            categoryPojo.setTotalPrice(roundTotalMoney);
            //收集数据
            out.collect(categoryPojo);
        }
    }

    /**
     * WindowResultProcess实现ProcessWindowFunction
     * abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window>
     * abstract class ProcessWindowFunction<CategoryPojo, Object, Tuple就是String类型的dateTime, TimeWindow extends Window>
     * <p>
     * 把各个分类的总价加起来，就是全站的总销量金额，
     * 然后我们同时使用优先级队列计算出分类销售的Top3，peek:展示 poll:移除
     * 最后打印出结果，在实际中我们可以把这个结果数据存储到hbase或者redis中，以供前端的实时页面展示。
     */
    private static class WindowResultProcess extends ProcessWindowFunction<CategoryPojo, Object, String, TimeWindow> {
        @Override
        public void process(String datetime, Context context, Iterable<CategoryPojo> elements, Collector<Object> out) throws Exception {
            //1. 进行累加计算，得到每一秒的总金额
            double totalMoney = 0D;
            double roundTotalMoney = 0D;
            //2. 计算分类的前三名 使用小顶堆进行排序 找出前三名
            //PriorityBlockingQueue 线程安全的堆的队列 自动排序
            //时间复杂度 - O(1)  //空间复杂度 - O(1)
            PriorityBlockingQueue<CategoryPojo> queues = new PriorityBlockingQueue<>(
                    3,
                    (o1, o2) -> o1.totalPrice >= o2.totalPrice ? 1 : -1
            );

            for (CategoryPojo element : elements) {
                totalMoney += element.totalPrice;

                //往队列中插入数据
                /**
                 * 20
                 * 30      =>    45
                 * 55
                 *
                 * 30
                 * 45      =>    65
                 * 55
                 *
                 * 45
                 * 55
                 * 65
                 *
                 */
                if (queues.size() < 3) {
                    queues.add(element);
                } else {
                    CategoryPojo top = queues.peek();
                    if (top.totalPrice < element.totalPrice) {
                        queues.poll();
                        queues.add(element);
                    }
                }
            }

            BigDecimal bigDecimal = new BigDecimal(totalMoney);
            roundTotalMoney = bigDecimal.setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();

            List<CategoryPojo> sortedList = queues.stream()
                    .sorted((o1, o2) -> o1.totalPrice > o2.totalPrice ? -1 : 1)
                    .collect(Collectors.toList());

            System.out.println("时间 ： " + datetime + "  总价 : " + roundTotalMoney + " top3:\n" + StringUtils
                    .join(sortedList, ",\n"));
            System.out.println("-------------");
        }
    }
}
